package cn.zwx.canal.tools.scheduling;

import cn.zwx.canal.tools.event.AlterAbstractCanalEvent;
import cn.zwx.canal.tools.event.DeleteAbstractCanalEvent;
import cn.zwx.canal.tools.event.InsertAbstractCanalEvent;
import cn.zwx.canal.tools.event.UpdateAbstractCanalEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @description: 通过定时任务去定时拉取
 * @projectName:learn
 * @see:cn.zwx.canal.tools.scheduling
 * @author:zhangwenxue
 * @createTime:2020/12/19 10:47
 * @version:1.0
 */
@Component
public class CanalScheduling implements Runnable, ApplicationContextAware {

    private static final Logger logger = LoggerFactory.getLogger(CanalScheduling.class);

    private ApplicationContext applicationContext;

    @Value("${canal.batch.size}")
    private Integer batchSize;

    @Autowired
    private CanalConnector canalConnector;

    @Scheduled(cron = "0/10 * * * * ? ")
    @Override
    public void run() {
        try {
            Message message = canalConnector.getWithoutAck(batchSize);
            long batchId = message.getId();
            logger.info("通过定时任务去定时拉取,获取到message:",JSON.toJSONString(message));
            try {
                List<CanalEntry.Entry> entries = message.getEntries();
                if (batchId != -1 && entries.size() > 0) {
                    entries.forEach(entry -> {
                        if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                            logger.info("publishCanalEvent=" + entry.getEntryType());
                            logger.info("message=" + message.toString());
                            this.publishCanalEvent(entry);
                        }
                    });
                }
                canalConnector.ack(batchId);
            } catch (Exception e) {
                logger.error("发送监听事件失败！batchId回滚,batchId=" + batchId, e);
                canalConnector.rollback(batchId);
            }
        }catch (Exception e){
            logger.error("canal_scheduled异常！", e);
        }
    }

    /**
     * @description 根据SQL请求去构建不同的事件
     * @return void
     * @author zhangwenxue
     * @createTime 2020/12/19 15:03
     **/
    private void publishCanalEvent(CanalEntry.Entry entry) {
        CanalEntry.EventType eventType = entry.getHeader().getEventType();
        switch (eventType) {
            case INSERT:
                applicationContext.publishEvent(new InsertAbstractCanalEvent(entry));
                break;
            case UPDATE:
                applicationContext.publishEvent(new UpdateAbstractCanalEvent(entry));
                break;
            case DELETE:
                applicationContext.publishEvent(new DeleteAbstractCanalEvent(entry));
                break;
            case ALTER:
                applicationContext.publishEvent(new AlterAbstractCanalEvent(entry));
                break;
            default:
                break;
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
